package com.butterfly.hotkey.client.core.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.thread.NamedThreadFactory;
import com.alibaba.fastjson2.JSON;
import com.butterfly.hotkey.client.core.collector.IKeyCollector;
import com.butterfly.hotkey.client.core.config.AppClientProperties;
import com.butterfly.hotkey.core.constant.MsgTypeConst;
import com.butterfly.hotkey.core.constant.NumConst;
import com.butterfly.hotkey.core.model.HotKeyModel;
import com.butterfly.hotkey.core.model.KeyCountModel;
import com.butterfly.hotkey.core.model.KeyModel;
import com.butterfly.hotkey.core.model.msg.HotKeyMsg;
import com.butterfly.hotkey.core.util.NettyMsgUtil;
import com.google.common.collect.Maps;
import io.netty.channel.Channel;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 推送的定时器
 *
 * @author butterfly
 * @since 2024/11/2
 */
public class PushScheduleService {

    private final KeyCollectorService keyCollectorService;
    private final AppClientProperties clientProperties;
    private final WorkerService workerService;

    public PushScheduleService(KeyCollectorService keyCollectorService,
                               WorkerService workerService,
                               AppClientProperties clientProperties) {
        this.keyCollectorService = keyCollectorService;
        this.workerService = workerService;
        this.clientProperties = clientProperties;

        this.schedulePushKeyAccess();
        this.schedulePushSummaryKey();
    }

    /**
     * 推送一次待测key到worker， 越快检测越即时
     */
    public void schedulePushKeyAccess() {
        // 最小200ms
        long millis = clientProperties.getPushKeyIntervalMillis();
        if (millis < NumConst.TWO_HUNDRED) {
            millis = NumConst.TWO_HUNDRED;
        }
        NamedThreadFactory threadFactory = new NamedThreadFactory("hotkey-key-pusher-executor", true);
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(threadFactory);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            IKeyCollector<HotKeyModel, HotKeyModel> keyAccessCollector = keyCollectorService.getKeyAccessCollector();
            Collection<HotKeyModel> keys = keyAccessCollector.getCollection();
            if (CollectionUtil.isNotEmpty(keys)) {
                pushKey(keys, clientProperties.getAppName());
            }
        }, 0, millis, TimeUnit.MILLISECONDS);
    }

    /**
     * 推送一次key的访问量到worker
     * worker之后再推送给dashboard进行统计
     */
    public void schedulePushSummaryKey() {
        Long pushSummarySecond = clientProperties.getPushSummarySecs();

        NamedThreadFactory threadFactory = new NamedThreadFactory("hotkey-summary-pusher-executor", true);
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(threadFactory);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            IKeyCollector<KeyModel, KeyCountModel> keyAccessCollector = keyCollectorService.getKeySummaryCollector();
            Collection<KeyCountModel> keys = keyAccessCollector.getCollection();
            if (CollectionUtil.isNotEmpty(keys)) {
                pushKeyCount(keys, clientProperties.getAppName());
            }
        }, 0, pushSummarySecond, TimeUnit.SECONDS);
    }

    private void pushKeyCount(Collection<KeyCountModel> hotkeys, String appName) {
        Map<Channel, List<KeyCountModel>> keyMap = Maps.newHashMapWithExpectedSize(hotkeys.size());
        long now = SystemClock.now();
        for (KeyCountModel hotkey : hotkeys) {
            Channel channel = workerService.chooseChannel(hotkey.getRuleKey());
            if (channel == null) {
                continue;
            }
            List<KeyCountModel> newList = keyMap.computeIfAbsent(channel, k -> new ArrayList<>());
            hotkey.setCreateTime(now);
            newList.add(hotkey);
        }

        // 按照不同分片进行处理
        for (Map.Entry<Channel, List<KeyCountModel>> entry : keyMap.entrySet()) {
            Channel channel = entry.getKey();
            List<KeyCountModel> models = entry.getValue();
            HotKeyMsg hotKeyMsg = new HotKeyMsg(MsgTypeConst.REQUEST_HIT_COUNT);
            hotKeyMsg.setAppName(appName);
            hotKeyMsg.setBody(JSON.toJSONString(models));

            channel.writeAndFlush(NettyMsgUtil.buildByteBuf(hotKeyMsg));
        }
    }

    private void pushKey(Collection<HotKeyModel> hotkeys, String appName) {
        long now = SystemClock.now();

        Map<Channel, List<HotKeyModel>> keyMap = Maps.newHashMapWithExpectedSize(hotkeys.size());
        for (HotKeyModel hotkey : hotkeys) {
            hotkey.setCreateTime(now);
            Channel channel = workerService.chooseChannel(hotkey.getKey());
            if (channel == null) {
                continue;
            }

            List<HotKeyModel> newList = keyMap.computeIfAbsent(channel, k -> new ArrayList<>());
            newList.add(hotkey);
        }

        // 按照不同分片进行处理
        for (Map.Entry<Channel, List<HotKeyModel>> entry : keyMap.entrySet()) {
            Channel channel = entry.getKey();
            List<HotKeyModel> models = entry.getValue();
            // 推送给worker
            HotKeyMsg hotKeyMsg = new HotKeyMsg(MsgTypeConst.REQUEST_NEW_KEY);
            hotKeyMsg.setAppName(appName);
            hotKeyMsg.setHotKeyModels(models);

            channel.writeAndFlush(NettyMsgUtil.buildByteBuf(hotKeyMsg));
        }
    }
}
